package com.guaoran.micro.services.spring.cloud.stream.kafka.api;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author : gucheng
 * @Description : <br/>
 *  kafka 生产者
 * @date :2018/12/6 22:43
 */
public class KafkaProducerDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 初始化配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.45.135:9092");
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());
        // 创建 Kafka Producer
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
        // 创建 Kakfa 消息  = ProducerRecord
        String topic = "guaoran";
        Integer partition = 0;
        Long timestamp = System.currentTimeMillis();
        String key = "message-key";
        String value = "孤傲然 ";
        ProducerRecord<String, String> record =
                new ProducerRecord<String, String>(topic, partition, timestamp, key, value);
        // 发放 Kakfa 消息
        Future<RecordMetadata> metadataFuture =  kafkaProducer.send(record);
        // 强制执行
        metadataFuture.get();
    }
}
